package org.qiyu.live.user.provider.config;

import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import org.idea.qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import org.qiyu.live.user.constants.UserProviderTopicNames;
import org.qiyu.live.user.dto.UserCacheAsyncDeleteDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;

import static org.qiyu.live.user.constants.CacheAsyncDeleteCode.USER_INFO_DELETE;
import static org.qiyu.live.user.constants.CacheAsyncDeleteCode.USER_TAG_DELETE;

/**
 * RocketMQ的消费者bean配置类
 */
@Configuration
public class RocketMQConsumerConfig implements InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQConsumerConfig.class);

    // 消费者的属性
    @Resource
    private RocketMQConsumerProperties rocketMQConsumerProperties;
    // 用于->延迟消息->删除缓存
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    // 构建redisBuilder->构建key
    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;

    // 容器初始化的时候,就初始化消费者
    @Override
    public void afterPropertiesSet() throws Exception {
        initConsumer();
    }

    public void initConsumer() {
        try {
            DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer();
            mqPushConsumer.setVipChannelEnabled(false);
            mqPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
            mqPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName() + "_" + RocketMQConsumerConfig.class.getSimpleName());
            mqPushConsumer.setConsumeMessageBatchMaxSize(1);
            mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            mqPushConsumer.subscribe(UserProviderTopicNames.CACHE_ASYNC_DELETE_TOPIC, "");
            // 消费消息->监听器
            mqPushConsumer.setMessageListener(new MessageListenerConcurrently() {
                // 消费消息
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    String messageJSON = new String(msgs.get(0).getBody());
                    // 解析出对象
                    UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = JSON.parseObject(messageJSON, UserCacheAsyncDeleteDTO.class);
                    if (userCacheAsyncDeleteDTO == null) {
                        LOGGER.error("用户id为空，参数异常，内容:{}", messageJSON);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    if (USER_INFO_DELETE.getCode() == userCacheAsyncDeleteDTO.getCode()) {
                        Long userId = JSON.parseObject(userCacheAsyncDeleteDTO.getJson(), long.class);
                        if (userId == null) {
                            LOGGER.error("用户id为空，参数异常，内容:{}", messageJSON);
                        }
                        String key = cacheKeyBuilder.buildUserInfoKey(userId);
                        redisTemplate.delete(key);
                    } else if (USER_TAG_DELETE.getCode() == userCacheAsyncDeleteDTO.getCode()) {
                        Long userId = JSON.parseObject(userCacheAsyncDeleteDTO.getJson(), long.class);
                        if (userId == null) {
                            LOGGER.error("用户id为空，参数异常，内容:{}", messageJSON);
                        }
                        String key = cacheKeyBuilder.buildUserTagKey(userId);
                        redisTemplate.delete(key);
                    }
                    LOGGER.error("延迟删除处理，userCacheAsyncDeleteDTO is {}", userCacheAsyncDeleteDTO);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者
            mqPushConsumer.start();
            LOGGER.info("mq消费者启动成功,namesrv is {}", rocketMQConsumerProperties.getNameSrv());
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
    }
}